/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;

import taskgraph.ports.InputPort;
import taskgraph.ports.OutputPort;
import taskgraph.tasks.PrimitiveTask;


/**
 * Sums the values received in wrapper objects.
 * 
 * <p>This is a generic task that will process wrapped values of types byte,
 * short, int, long, float, and double. Thus the input port expects Object
 * references of types Byte, Short, Integer, Long, Float, and Double,
 * respectively. The output channel will transport the sum total as a wrapped 
 * value of the same type as the input.
 * 
 * <p>This scheme is intended to be compatible with tasks Split and Compose.
 * For better performance, custom tasks may be written using channels that
 * transport native values.
 * 
 * @author Armando Blancas
 * @see taskgraph.ports.InputPort
 * @see taskgraph.ports.OutputPort
 */
public class Sum extends PrimitiveTask {

	private InputPort<Object> input;
	private OutputPort<Object> output;
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public Sum() {
    	// do nothing
    }

	/**
     * Sum constructor.
     * 
     * @param input The input port.
     * @param output The output port.
     */
    public Sum(final InputPort<Object> input, final OutputPort<Object> output) {
    	setInput(input);
    	setOutput(output);
    }

    /**
     * Gets the input port.
     * 
	 * @return The input port.
	 */
	public InputPort<Object> getInput() {
		return input;
	}
	
    /**
     * Gets the output port.
     * 
	 * @return The output port.
	 */
	public OutputPort<Object> getOutput() {
		return output;
	}

	/**
     * Performs the work of this task.
     * 
	 *<pre>
	 *while not EOF
	 *    read an object reference from input port
	 *    switch on type
	 *        loop through and add up input
	 *close input port
	 *write Sum to output port
	 *close output port
	 *</pre>
	 */
	@Override
    public void run() {
    	try {
    		Object obj = input.read();
    		if (obj instanceof Byte) {
    			sumBytes((Byte) obj);
    		} else if (obj instanceof Short) {
    			sumShorts((Short) obj);
    		} else if (obj instanceof Integer) {
    			sumIntegers((Integer) obj);
    		} else if (obj instanceof Long) {
    			sumLongs((Long) obj);
    		} else if (obj instanceof Float) {
    			sumFloats((Float) obj);
    		} else if (obj instanceof Double) {
    			sumDoubles((Double) obj);
    		}
        } catch (InterruptedIOException ie) {
        	// interruption request
        } catch (EOFException e) {
        	// we're done
        } catch (IOException e) {
        	// pipe synch error
            log.error("Sum#run()", e);
		} finally {
			input.close();
			output.close();
        }
    }
	
	/**
	 * Sets the input port.
	 * 
	 * @param input The input port.
	 */
    public void setInput(final InputPort<Object> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}
	
	/**
	 * Sets the output port.
	 * 
	 * @param output The output port.
	 */
    public void setOutput(final OutputPort<Object> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}
	
	/**
	 * Perform the sum for wrapped {@code byte} values.
	 * 
	 * @param initial The first number read from caller method.
	 * @throws IOException If there is any trouble with the input port.
	 */
	private void sumBytes(byte initial) throws IOException {
		byte sum = initial;
		while (!Thread.currentThread().isInterrupted()) {
			try {
				sum += (Byte) input.read();
			}
			catch (EOFException e) {
				break;
			}
        }
		output.write(sum);
	}
	
	/**
	 * Perform the sum for wrapped {@code double} values.
	 * 
	 * @param initial The first number read from caller method.
	 * @throws IOException If there is any trouble with the input port.
	 */
	private void sumDoubles(double initial) throws IOException {
		double sum = initial;
		while (!Thread.currentThread().isInterrupted()) {
			try {
				sum += (Double) input.read();
			}
			catch (EOFException e) {
				break;
			}
        }
		output.write(sum);
	}
	
	/**
	 * Perform the sum for wrapped {@code float} values.
	 * 
	 * @param initial The first number read from caller method.
	 * @throws IOException If there is any trouble with the input port.
	 */
	private void sumFloats(float initial) throws IOException {
		float sum = initial;
		while (!Thread.currentThread().isInterrupted()) {
			try {
				sum += (Float) input.read();
			}
			catch (EOFException e) {
				break;
			}
        }
		output.write(sum);
	}
	
	/**
	 * Perform the sum for wrapped {@code int} values.
	 * 
	 * @param initial The first number read from caller method.
	 * @throws IOException If there is any trouble with the input port.
	 */
	private void sumIntegers(int initial) throws IOException {
		int sum = initial;
		while (!Thread.currentThread().isInterrupted()) {
			try {
				sum += (Integer) input.read();
			}
			catch (EOFException e) {
				break;
			}
        }
		output.write(sum);
	}

	/**
	 * Perform the sum for wrapped {@code long} values.
	 * 
	 * @param initial The first number read from caller method.
	 * @throws IOException If there is any trouble with the input port.
	 */
	private void sumLongs(long initial) throws IOException {
		long sum = initial;
		while (!Thread.currentThread().isInterrupted()) {
			try {
				sum += (Long) input.read();
			}
			catch (EOFException e) {
				break;
			}
        }
		output.write(sum);
	}

	/**
	 * Perform the sum for wrapped {@code short} values.
	 * 
	 * @param initial The first number read from caller method.
	 * @throws IOException If there is any trouble with the input port.
	 */
	private void sumShorts(short initial) throws IOException {
		short sum = initial;
		while (!Thread.currentThread().isInterrupted()) {
			try {
				sum += (Short) input.read();
			}
			catch (EOFException e) {
				break;
			}
        }
		output.write(sum);
	}

}
